/**
 * File: MaqReadProcessor.java
 * Created by: mhaimel
 * Created on: 7 Jul 2009
 * CVS:  $Id: MaqReadProcessor.java,v 1.3 2009/12/07 10:02:21 mhaimel Exp $
 */
package uk.ac.ebi.curtain.processor.maq.impl;

import java.io.BufferedReader;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import org.apache.commons.lang.NotImplementedException;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import uk.ac.ebi.curtain.controller.maq.MaqBean;
import uk.ac.ebi.curtain.processor.AbstractProcessor;
import uk.ac.ebi.curtain.util.file.FileChannelAccess;
import uk.ac.ebi.curtain.utils.CurtainUncheckedException;
import uk.ac.ebi.curtain.utils.data.FileType;
import uk.ac.ebi.curtain.utils.data.ReadType;
import uk.ac.ebi.curtain.utils.data.impl.MaqIdentifierProcessor;
import uk.ac.ebi.curtain.utils.data.impl.SolexaIdentifierProcessor;
import uk.ac.ebi.curtain.utils.file.FileInfo;
import uk.ac.ebi.curtain.utils.file.SplitUtil;
import uk.ac.ebi.curtain.utils.file.impl.RollingFileSplitProcessor;
import uk.ac.ebi.curtain.utils.io.impl.FileIO;

/**
 * @author mhaimel
 *
 */
public class MaqReadProcessor extends  AbstractProcessor<MaqBean, MaqBean> {
	private static final long _MIN_FILE_BYTE_SIZE = 1000000L; // not quite 1 MB
	private Log log = LogFactory.getLog(this.getClass());

	@Override
	protected MaqBean buildReturnObject() {
		return this.getInput();
	}

	@Override
	public void process() {
		File baseDir = getInput().getReadDir();
		for(FileInfo fi : getInput().getReadFiles()){
			List<FileInfo> splitFastqFiles = getFastqFiles(baseDir, fi);
			getInput().getReadBfqList().addAll(splitFastqFiles);
		}
	}

	public List<FileInfo> getFastqFiles(File baseDir, FileInfo fastq) {
		List<FileInfo> files = new ArrayList<FileInfo>();
		if(fastq.getFileType().equals(FileType.fasta)){
			FileInfo out = new FileInfo(
					new File(baseDir,
							FileType.fastq.addPostfix(fastq.getNameWithoutFilePostfix())),
					FileType.fastq,
					fastq.getReadType());
			transform(fastq, out);
			fastq = out;
		} else if(! fastq.getFileType().equals(FileType.fastq)) {
			// anything else than fastq
			throw new NotImplementedException(
					"Read files only supported in fasta and fastq file format");
		}
		if(isSplitRequired(fastq)){
			if(isReadPaired(fastq)){
				files.addAll(splitFastq(baseDir,fastq,2,true));
				// even number for paired reads
				if(files.size()%2 != 0){
					throw new CurtainUncheckedException(
							"Even amount of files expected for paired reads: " 
							+ files.size() + " files: " + StringUtils.join(files,','));
				}
			} else {
				files.addAll(splitFastq(baseDir,fastq,1,false));
			}
		} else {
			files.add(fastq);
		}
		return files;
	}
	
	private void transform(FileInfo read, FileInfo out) {
		getLog().debug("Transform file "+read + " --> " + out);
		try {
			FileType.transform(read, out);
		} catch (IOException e) {
			throw new CurtainUncheckedException(
					"Problems transforming Read: " + read + " --> " + out);
		}
	}
	
	private Collection<? extends FileInfo> splitFastq(File baseDir, FileInfo fastq, int nFiles, boolean idConv) {
		getLog().debug("Split read file into " + nFiles + " subfiles: " + fastq);
		SplitUtil splitter = fastq.getFileType().getSplitter();
		long maxSize = fastq.getFile().length();
		if(hasMaxFileSize()){
			getLog().debug("Set max file size to " + getMaxFileSize() + "!");
			maxSize = getMaxFileSize();
		}
		RollingFileSplitProcessor proc = new RollingFileSplitProcessor(
				fastq.getFileType(), 
				baseDir,
				"split-reads",
				nFiles,
				maxSize);
		FileChannelAccess<BufferedReader> inAcc = null;
		try{
			inAcc = getContext().getFileAccess().getBufferedReader(fastq.getFile());
			inAcc.open();
			if(idConv){
				splitter.split(new SolexaIdentifierProcessor(),new MaqIdentifierProcessor(), inAcc.getChannel(), proc);
			} else {
				splitter.split(null,null, inAcc.getChannel(), proc);
			}
		} catch (Exception e) {
			throw new CurtainUncheckedException("Problems while splitting file " + fastq,e);
		} finally{
			FileIO.closeQuietly(proc);
			FileIO.closeQuietly(inAcc);
		}
		log.debug("Finished splitting " + fastq + " into " + StringUtils.join(proc.getFileList(),','));
		List<FileInfo> fastqList = wrap(fastq.getReadType(), fastq.getFileType(), proc.getFileList());
		return fastqList;
	}
	
	private List<FileInfo> wrap(ReadType rType,FileType fType,Iterable<File> fileColl){
		List<FileInfo> fileList = new ArrayList<FileInfo>();
		for(File f : fileColl){
			fileList.add(new FileInfo(f,fType,rType));
		}
		return fileList;
	}
	
	private boolean isSplitRequired(FileInfo fastq) {
		return isReadPaired(fastq) || (hasMaxFileSize() && fastq.getFile().length() > getMaxFileSize()); // > 1GB (1024*1024*1024)
	}
	
	
	protected boolean hasMaxFileSize(){
		return getMaxFileSize() > _MIN_FILE_BYTE_SIZE;
	}
	
	protected long getMaxFileSize() {
		return getConfig().getMaqMaxFastqFile();
	}

	private boolean isReadPaired(FileInfo fastq){
		return fastq.getReadType().isPaired();
	}	
}
